package cn.lzd

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by lzd on 2018/12/21.
  */
object GXvariable {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("variable"))
    demo2(sc)
  }

  /**
    * Accumulator
    * Spark提供的Accumulator，主要用于多个节点对一个变量进行共享的操作，task只能对Accumulator进行累加操作，不能读取它的值，只有Driver程序可以读取。
    * https://blog.csdn.net/u013468917/article/details/70617085
    * @param sc
    */
  def demo2(sc: SparkContext): Unit = {
    val acc = sc.longAccumulator("hello")

    val seq = Array[Long](1,2,3,4,5,6,7,8,9,10)
    val rdd1 = sc.parallelize(seq)

    rdd1.foreach(v =>acc.add(v))

    println(acc.value)

  }

  /**
    * Broadcast Variable（广播变量）
    * Spark提供的Broadcast Variable是只读的，可以通过调用SparkContext的broadcast()方法来针对某个变量创建广播变量。每个节点可以使用广播变量的value()方法来获取值。
    *
    * @param sc
    */
  def demo1(sc: SparkContext): Unit = {
    val lst = Map("hello" -> "ok", "spark" -> "ok", "hadoop" -> "ok")
    val factorBroadcast = sc.broadcast(lst)

    val lst2 = Seq("hello")
    val rdd1 = sc.parallelize(lst2)


    rdd1.foreach(rdd => {
      factorBroadcast.value.get(rdd) match {
        case Some(value) => println(value)
        case None => println("no")
      }
    })
  }
}
